package com.sunday.common.mq.kafka.study.spring.e2_Configuring_Topics;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.config.TopicConfig;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

@Slf4j
@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        ConfigurableApplicationContext applicationContext = SpringApplication.run(Application.class, args);
        KafkaAdmin kafkaAdmin = applicationContext.getBean(KafkaAdmin.class);
        kafkaAdmin.setCreateOrModifyTopic(newTopic -> newTopic.name().equals("defaultBoth"));
    }

    @Bean
    public KafkaAdmin admin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        return new KafkaAdmin(configs);
    }

    @Bean
    public KafkaAdmin.NewTopics topics456() {
        return new KafkaAdmin.NewTopics(
                TopicBuilder.name("defaultBoth")
                        .build(),
                TopicBuilder.name("defaultPart")
                        .replicas(1)
                        .build(),
                TopicBuilder.name("defaultRepl")
                        .partitions(3)
                        .build());
    }

//    @Bean
    public NewTopic topic1() {
        return TopicBuilder.name("thing1")
                /**
                 * offsets.topic.num.partitions
                 * 偏移提交主题的分区数（部署后不应更改）。
                 */
                .partitions(5)
                /**
                 * 这里重点说一下，如果没有
                 *
                 * https://kafka.apache.org/documentation/#brokerconfigs_offsets.topic.replication.factor
                 * 在 Kafka 中，offsets.topic.replication.factor 是一个重要的配置属性，用于指定 Kafka 内部存储消费者偏移量（consumer offsets）的主题的副本因子（replication factor）。
                 * Kafka 使用内部的 __consumer_offsets 主题来存储和跟踪消费者组和其消费者偏移量的信息。这个主题是由 Kafka 自动创建和管理的，并且有自己的副本因子设置。
                 * offsets.topic.replication.factor 属性允许您指定 __consumer_offsets 主题的副本因子。副本因子决定了该主题在 Kafka 集群中的可靠性和容错能力。
                 * 副本因子定义了在集群中复制 __consumer_offsets 主题的副本数量。例如，如果将 offsets.topic.replication.factor 设置为 3，则表示 __consumer_offsets 主题会被复制成三个副本，分布在 Kafka 集群中的不同节点上。
                 * 通过设置适当的副本因子，可以提高数据的可靠性和可用性。如果某个 Kafka 节点故障或不可用，仍然可以从其他节点复制的副本中读取并恢复 __consumer_offsets 主题的数据。
                 * 需要注意的是，副本因子不能大于可用的 Kafka 副本数。确保为 offsets.topic.replication.factor 设置一个合理的值，以便充分利用可用的 Kafka 副本来确保数据的持久性和可靠性。
                 * 综上所述，offsets.topic.replication.factor 是一个配置属性，用于指定 Kafka 内部存储消费者偏移量的主题 __consumer_offsets 的副本因子。它决定了该主题在 Kafka 集群中的复制数量，以提供数据的可靠性和容错能力。
                 */
                .replicas(3)
                /**
                 * 清理策略
                 * 此配置指定要在日志段上使用的保留策略。“删除”策略（默认设置）将在达到保留时间或大小限制时丢弃旧段。
                 * “压缩”策略将启用日志压缩，该压缩保留每个键的最新值。也可以在逗号分隔的列表中指定这两个策略（例如“删除，压缩”）。
                 * 在这种情况下，将根据保留时间和大小配置丢弃旧段，而保留的段将被压缩。
                 */
                .compact()
                .build();
    }

//    @Bean
    public NewTopic topic2() {
        return TopicBuilder.name("thing2")
                .partitions(4)
                .replicas(2)
                /**
                 * 压缩类型。 compression.type
                 * 指定给定主题的最终压缩类型。
                 * 此配置接受标准压缩编解码器（“gzip”、“snappy”、“lz4”、“zstd”）。
                 * 它还接受“未压缩”，相当于无压缩;和“生产者”，这意味着保留生产者设置的原始压缩编解码器。
                 */
                .config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
                .build();
    }

//    @Bean
    public NewTopic topic3() {
        return TopicBuilder.name("thing3")
                .assignReplicas(0, Arrays.asList(0, 1))
                .assignReplicas(1, Arrays.asList(1, 2))
                .assignReplicas(2, Arrays.asList(2, 0))
                .config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
                .build();
    }

}
